package com.czhi.multi;

import com.lmax.disruptor.*;
import com.lmax.disruptor.dsl.ProducerType;

import java.util.UUID;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class Main {
    public static void main(String[] args) {

        //1. 创建RingBuffer
        RingBuffer<Order> ringBuffer = RingBuffer.create(ProducerType.MULTI,
                () -> new Order(), 1024 * 1024, new YieldingWaitStrategy());
        //2. 通过ringBuffer 创建一个屏障
        final SequenceBarrier sequenceBarrier = ringBuffer.newBarrier();
        //3. 构建多消费者
        Consumer[] consumers = new Consumer[10];
        for (int i = 0; i < 10; i++) {
            consumers[i] = new Consumer("C"+i);
        }
        //4. 构建多消费者工作池
        WorkerPool<Order> workerPool = new WorkerPool<>(ringBuffer, sequenceBarrier, new ExceptionHandler<Order>() {
            @Override
            public void handleEventException(Throwable ex, long sequence, Order event) {

            }

            @Override
            public void handleOnStartException(Throwable ex) {

            }

            @Override
            public void handleOnShutdownException(Throwable ex) {

            }
        }, consumers);

        //5. 设置多个消费者的sequence序号用于单独统计消费进度
        ringBuffer.addGatingSequences(workerPool.getWorkerSequences());

        // 6. 启动workPool
        final ExecutorService threadPool = Executors.newFixedThreadPool(2);
        workerPool.start(threadPool);
        CountDownLatch latch = new CountDownLatch(1);
        for (int i = 0; i < 100; i++) {
            Producer producer = new Producer(ringBuffer);
            new Thread(()->{
                try {
                    latch.await();
                } catch (Exception e) {
                    e.printStackTrace();
                }
                for (int j = 0; j < 100; j++) {
                    producer.sentData(UUID.randomUUID().toString());
                }
            }).start();
        }
        System.out.println("线程创建完毕");
        latch.countDown();
        threadPool.shutdown();

    }
}
